package com.catmiao.sql;


import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.expressions.Aggregator;
import scala.Serializable;

// TODO 自定义UDAF函数，实现年龄的平均值
//     1. 创建自定义的类
//     2. 继承Aggregator类
//     3. 设定泛型
//          IN： 输入数据类型
//          BUFF 缓冲区类型
//          OUT：输出数据类型
//      4. 重写方法 (4 +2 )
public class MyAvgAgeUDAF  extends Aggregator<Long,AvgAgeBuffer,Long> {

    // 初始化 缓冲区初始化
    @Override
    public AvgAgeBuffer zero() {
        return new AvgAgeBuffer(0L,0L);
    }


    // 聚合操作
    @Override
    public AvgAgeBuffer reduce(AvgAgeBuffer b, Long a) {
        b.setCount(b.getCount()+1);
        b.setTotal(b.getTotal() + a);
        return b;
    }

    // 合并缓冲区
    @Override
    public AvgAgeBuffer merge(AvgAgeBuffer b1, AvgAgeBuffer b2) {
        b1.setCount(b1.getCount() + b2.getCount());
        b1.setTotal(b1.getTotal() + b2.getTotal());
        return b1;
    }


    // 计算结果
    @Override
    public Long finish(AvgAgeBuffer reduction) {
        return reduction.getTotal() / reduction.getCount();
    }

    @Override
    public Encoder<AvgAgeBuffer> bufferEncoder() {
        return Encoders.bean(AvgAgeBuffer.class);
    }

    @Override
    public Encoder<Long> outputEncoder() {
        return Encoders.LONG();
    }
}

